package com.wyh.apitest.sink

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.wyh.apitest.source.SensorReading
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests

/**
  *
  * Flink写入数据到ES
  */
object EsSinkDemo {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val df = new SimpleDateFormat()

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")
    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })


    //定义httpHosts ES连接节点列表List
    val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("localhost", 9200))

    //定义写入ES的EsSinkFunction
    val MyEsSinkFunction = new ElasticsearchSinkFunction[SensorReading] {
      //每条数据都会执行这个方法
      override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {

        val date1 = new Date()

        val date: String = df.format(date1)

        //包装一个Map作为data source
        val dataSource = new util.HashMap[String, String]()
        dataSource.put("id", t.id)
        dataSource.put("temperature", t.temperature.toString)
        dataSource.put("ts", date)

        //创建indexRequest用于发送Http请求
        val indexRequest = Requests.indexRequest()
          .index("sensor")
          .`type`("readingdata")
          .source(dataSource)

        //用indexer发送请求
        requestIndexer.add(indexRequest)

      }
    }

    dataStream.addSink(new ElasticsearchSink
    .Builder[SensorReading](httpHosts, MyEsSinkFunction)
      .build()
    )


    streamEnv.execute("Elasticsearch Sink Demo")

  }
}